
package com.hlkj.pay.listener;
/*
 * Hlpay-Plus aggregate payment system.
 * Copyright (c) 2024-2025 Hlpay Team Copyright has the right of final interpretation.
 */

import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.hlkj.framework.caches.service.RedisTemplateService;
import com.hlkj.pay.app.order.IMerchantPaymentStatisticsAppService;
import com.hlkj.pay.common.constants.AdRedisConstant;
import com.hlkj.pay.common.constants.MqConstant;
import com.hlkj.pay.dto.payment.MerchantPaymentDetailDto;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;

/**
 * @author HlpayTeam
 * @date 2024/10/24 12:58
 */
@Slf4j
@Component
@ConditionalOnProperty(prefix = "rocketmq", name = "enabled", havingValue = "true")
public class MerchantPaymentBatchConsumer {

    @Autowired
    private IMerchantPaymentStatisticsAppService paymentStatisticsAppService;

    @Value("${rocketmq.name-server}")
    private String nameServer;

    @Value("${rocketmq.consumer.pull-batch-size:400}")
    private Integer pullBatchSize;

    @Value("${rocketmq.consumer.consume-message-batch-max-size:400}")
    private Integer consumeMessageBatchMaxSize;

    @Value("${rocketmq.consumer.pull_interval:1000}")
    private Integer pullInterval;

    @Value("${rocketmq.consumer.consume_key_time:60}")
    private Integer consumeKeyTime;

    @Value("${server.port}")
    private String serverPort;

    @Autowired
    private RedisTemplateService redisTemplateService;

    @Bean
    public DefaultMQPushConsumer paymentMQPushConsumer() throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(MqConstant.PAYMENT_STATISTICS_TOPIC);
        consumer.setNamesrvAddr(nameServer);
        consumer.subscribe(MqConstant.PAYMENT_STATISTICS_TOPIC, "*");
        // 设置每次消息拉取的时间间隔，单位毫秒
        consumer.setPullInterval(pullInterval);
        // 设置每个队列每次拉取的最大消息数
        consumer.setPullBatchSize(pullBatchSize);
        consumer.setInstanceName(MqConstant.PAYMENT_STATISTICS_TOPIC + "_" + serverPort);
        // 设置消费者单次批量消费的消息数目上限
        consumer.setConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize);
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            log.info("merchant payment statistics receive messages: {}", msgs.size());
            List<MerchantPaymentDetailDto> statisticsList = new ArrayList<>(msgs.size());
            log.debug("merchant payment statistics receive topic: {}", msgs.get(0).getTopic());
            msgs.forEach(msg -> {
                MerchantPaymentDetailDto statisticsDto = JSONObject.parseObject(msg.getBody(), MerchantPaymentDetailDto.class);
                String lockKey = key(statisticsDto);
                Boolean hasKey = redisTemplateService.hasKey(lockKey);
                // 5分钟
                if (!hasKey) {
                    redisTemplateService.set(lockKey, statisticsDto.getDetailPaymentId(), consumeKeyTime);
                    statisticsList.add(statisticsDto);
                } else {
                    log.error("merchant payment statistics consumer lock fail orderNo:{} status:{}", statisticsDto.getDetailPaymentId(), statisticsDto.getStatus());
                }
            });
            log.info("merchant payment statisticsList size: {}", statisticsList.size());
            if (!CollectionUtils.isEmpty(statisticsList)) {
                // 处理批量消息
                paymentStatisticsAppService.processStatisticsMessage(statisticsList);
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
        return consumer;
    }

    String key(MerchantPaymentDetailDto statisticsDto) {
        return AdRedisConstant.AD_STATISTICS_CONSUMER_KEY + statisticsDto.getStatus() + ":" + statisticsDto.getDetailPaymentId();
    }
}
